package com.idanchuang.component.redis.listener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author yjy
 * Created at 2021/9/14 10:51 上午
 */
public class RedisKeyEventExpiredListener extends KeyExpirationEventMessageListener {

    private static final Logger logger = LoggerFactory.getLogger(RedisKeyEventExpiredListener.class);

    /** 串行 执行器 */
    final static ThreadPoolExecutor SERIAL_WORKER ;
    /** 并行执行器 */
    final static ThreadPoolExecutor CONCURRENT_WORKER;

    private final KeyExpiredProcessor processor;
    private final RedisProperties redisProperties;

    static {
        SERIAL_WORKER = new ThreadPoolExecutor(1, 1,
                1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(50000),
                (run) -> new Thread(run, "key-expire-worker-s"),
                new ThreadPoolExecutor.AbortPolicy());
        SERIAL_WORKER.allowCoreThreadTimeOut(true);

        int availableProcessors = Runtime.getRuntime().availableProcessors();
        CONCURRENT_WORKER = new ThreadPoolExecutor(availableProcessors, availableProcessors,
                1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(1000),
                (run) -> new Thread(run, "key-expire-worker-c"),
                new ThreadPoolExecutor.CallerRunsPolicy());
        CONCURRENT_WORKER.allowCoreThreadTimeOut(true);
    }

    public RedisKeyEventExpiredListener(RedisProperties redisProperties,
                                        RedisMessageListenerContainer listenerContainer,
                                        KeyExpiredProcessor processor) {
        super(listenerContainer);
        this.redisProperties = redisProperties;
        this.processor = processor;
    }

    @Override
    protected void doRegister(RedisMessageListenerContainer listenerContainer) {
        Topic topic = new PatternTopic("__keyevent@" + redisProperties.getDatabase() + "__:expired");
        listenerContainer.addMessageListener(this, topic);
    }

    @Override
    protected void doHandleMessage(Message message) {
        ExecutorService worker = processor.procMode() == ProcMode.SERIAL ? SERIAL_WORKER : CONCURRENT_WORKER;
        worker.execute(() -> {
            try {
                processor.process(message.getBody());
            } catch (Exception e) {
                logger.error("doHandleMessage error", e);
            }
        });
    }

}
